package cn.itcast.streaming.sink;

import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.utils.DateUtil;
import cn.itcast.streaming.utils.StringUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//车辆明细数据存入hbase表中
public class VehicleDetailSinkOptimizer extends RichSinkFunction<ItcastDataObj> {

    private final Logger logger = LoggerFactory.getLogger(VehicleDetailSinkOptimizer.class);
    private Connection conn;
    private String tableName;
    private BufferedMutator bufferedMutator;

    public VehicleDetailSinkOptimizer(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        this.conn = ConnectionFactory.createConnection(conf);
        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tableName));
        params.writeBufferSize(10*1024*1024);//指定内存中缓冲区大小
        this.bufferedMutator = conn.getBufferedMutator(params);
        System.out.println("连接hbase对象初始化完成");

    }

    @Override
    public void close() throws Exception {
        super.close();
        if(bufferedMutator!=null) bufferedMutator.close();
        if(conn!=null && !conn.isClosed()) conn.close();
    }

    @Override //将数据一条一条写入hbase
    public void invoke(ItcastDataObj value, Context context) throws Exception {

        //将itcastObj转换成put对象
        Put put = setDataSourcePut(value);
        bufferedMutator.mutate(put);
        bufferedMutator.flush();
    }

    private Put setDataSourcePut(ItcastDataObj itcastDataObj) {

        //电量百分比(currentElectricity)、
        // 当前电量(remainPower)、
        // 百公里油耗(fuelConsumption100km)、
        // 发动机速度(engineSpeed)、
        // 车辆速度(vehicleSpeed)
        // 车架号(vin)
        // 终端时间(terminalTime)
        // 存储时间(processTime)
        String rowkey = itcastDataObj.getVin()+ StringUtil.reverse(itcastDataObj.getTerminalTimeStamp().toString());

        Put put = new Put(Bytes.toBytes(rowkey));

        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("vin"), Bytes.toBytes(itcastDataObj.getVin()));
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("terminalTime"), Bytes.toBytes(itcastDataObj.getTerminalTime()));
        if(itcastDataObj.getCurrentElectricity()!=-999999D){//电量百分比
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("currentElectricity"), Bytes.toBytes(itcastDataObj.getCurrentElectricity()));
        }
        if(itcastDataObj.getRemainPower()!=-999999D){//当前电量
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("remainPower"), Bytes.toBytes(itcastDataObj.getRemainPower()));
        }
        if(itcastDataObj.getFuelConsumption100km()!=""){//百公里油耗
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("fuelConsumption100km"), Bytes.toBytes(itcastDataObj.getFuelConsumption100km()));
        }
        if(itcastDataObj.getEngineSpeed()!=""){//发动机速度
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("engineSpeed"), Bytes.toBytes(itcastDataObj.getEngineSpeed()));
        }
        if(itcastDataObj.getVehicleSpeed()!=-999999D){//车速
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("vehicleSpeed"), Bytes.toBytes(itcastDataObj.getVehicleSpeed()));
        }
        if(itcastDataObj.getCurrentElectricity()!=-999999D){
            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("currentElectricity"), Bytes.toBytes(itcastDataObj.getCurrentElectricity()));
        }
        //当前时间
        put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("processTime"),Bytes.toBytes(DateUtil.getCurrentTime()));

        return put;
    }
}
